"""
Case Type   : Audit
Case Name   : 兼容B库中执行dml语法审计，执行成功
Create At   : 2025/04/01
Owner       : liu-tong-8848
Description :
              1.审计检查执行dml语法
              sys.argv[1] = $GAUSSHOME/lib/postgresql/libog_query.so
              sys.argv[2] = "create table test_audit_001 ( id bigint(20) NOT NULL COMMENT '主键',
                             role_name varchar(64) NOT NULL, role_code varchar(64) NOT NULL COMMENT '角色组编码',
                             role_desc varchar(255) NULL DEFAULT NULL,
                             role_type tinyint(1) NULL DEFAULT NULL COMMENT '角色类型',
                             data_scope int(11) NOT NULL COMMENT '数据权限 1、本人;2、本人及下属;3、本部门;4、本部门及下属部门;
                             5、特定部门6、全部', sort int(11) NULL DEFAULT NULL COMMENT '排序',
                             status int(11) NULL DEFAULT 1 COMMENT '状态', create_by bigint(20) NOT NULL COMMENT '创建人',
                             update_by bigint(20) NOT NULL COMMENT '修改人', create_time datetime(0) NOT NULL
                             DEFAULT CURRENT_TIMESTAMP(0), update_time datetime(0) NOT NULL ON
                             UPDATE CURRENT_TIMESTAMP(0), del_flag char(1) NOT NULL DEFAULT '0' COMMENT '是否已删除',
                             PRIMARY KEY ( id ), UNIQUE INDEX role_idx1_role_code2 (role_code)) ENGINE = InnoDB
                             COMMENT = '系统角色表';
                             insert into test_audit_001(id, role_name, role_code, role_desc, role_type, data_scope,
                             status,create_by, update_by, create_time, update_time, del_flag) values (1, '超级管理员',
                             'ROLE_ADMIN', NULL, 1, 6, 1, 1, 1, '2021-12-17 00:00:00', '2021-12-17 00:00:00', '0');
                             update test_audit_001 set role_name='ado', role_type=1, sort=0,
                             update_time='2022-12-07 13:10:20', update_by=1 WHERE id=1 AND del_flag=0;
                             delete from test_audit_001;
                             select length1(a) as la from (SELECT deptno, AVG(sal) AS avgsal FROM emp GROUP BY deptno)
                             t2 where b in (select hex(t3.b) from t3 where c=10 having t3.a>1 order by t3.c limit 10);"
              # -*- coding: utf-8 -*-
              import sys
              import ctypes
              class OgQueryParseResult(ctypes.Structure):
                  _fields_ = [
                      ("parse_tree_json", ctypes.c_char_p),
                      ("err_msg", ctypes.c_char_p),
                      ("is_passed", ctypes.c_bool)
                  ]
              libpg_query = ctypes.CDLL(sys.argv[1])
              libpg_query.raw_parser_opengauss_dolphin.restype = OgQueryParseResult
              libpg_query.raw_parser_opengauss_dolphin.argtypes = [ctypes.c_char_p]
              sql_bytes = sys.argv[2].encode('utf-8')
              result = libpg_query.raw_parser_opengauss_dolphin(sql_bytes)
              print(sql_bytes)
              print(result.parse_tree_json)
              print(result.err_msg)
              print(result.is_passed)
Expect      :
              1.成功
History     :
"""

import unittest

from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger
from yat.test import Node
from yat.test import macro


class Audit(unittest.TestCase):

    def setUp(self):
        self.log = Logger()
        self.constant = Constant()
        self.log.info('==Opengauss_Function_Grammar_Audit_Case001开始==')
        self.python_file_name = "test_audit.py"
        self.db_user_node = Node(node='PrimaryDbUser')

    def test_add_dimension(self):
        text = '----step1: 审计检查审计检查执行dml语法 expect: 成功----'
        self.log.info(text)
        init_order_1 = "echo '" \
                       "# -*- coding: utf-8 -*-\n" \
                       "import sys\n" \
                       "import ctypes\n" \
                       "\n" \
                       "class OgQueryParseResult(ctypes.Structure):\n" \
                       "    _fields_ = [\n" \
                       "        (\"parse_tree_json\", ctypes.c_char_p),\n" \
                       "        (\"err_msg\", ctypes.c_char_p),\n" \
                       "        (\"is_passed\", ctypes.c_bool)\n" \
                       "    ]\n" \
                       "\n" \
                       "' > " + self.python_file_name
        init_order_2 = "echo \"" \
                       "libpg_query = ctypes.CDLL(sys.argv[1])\n" \
                       "libpg_query.raw_parser_opengauss_dolphin.restype = OgQueryParseResult\n" \
                       "libpg_query.raw_parser_opengauss_dolphin.argtypes = [ctypes.c_char_p]\n" \
                       "\n" \
                       "sql_bytes = sys.argv[2].encode('utf-8')\n" \
                       "result = libpg_query.raw_parser_opengauss_dolphin(sql_bytes)\n" \
                       "print(sql_bytes)\n" \
                       "print(result.parse_tree_json)\n" \
                       "print(result.err_msg)\n" \
                       "print(result.is_passed)\n" \
                       "\" >> " + self.python_file_name
        self.log.info(init_order_1)
        self.log.info(init_order_2)
        init_result_1 = self.db_user_node.sh(init_order_1).result()
        init_result_2 = self.db_user_node.sh(init_order_2).result()
        self.assertNotIn('ERROR', init_result_1, "执行失败")
        self.assertNotIn('ERROR', init_result_2, "执行失败")

        self.log.info("cat " + self.python_file_name)
        cat_result = self.db_user_node.sh("cat " + self.python_file_name).result()
        self.log.info(cat_result)

        test_so_file_name = "$GAUSSHOME/lib/postgresql/libog_query.so"
        test_sql = "create table test_audit_001 ( id bigint(20) NOT NULL COMMENT '主键', " \
                   + "role_name varchar(64) NOT NULL, role_code varchar(64) NOT NULL COMMENT '角色组编码', " \
                   + "role_desc varchar(255) NULL DEFAULT NULL, " \
                   + "role_type tinyint(1) NULL DEFAULT NULL COMMENT '角色类型', " \
                   + "data_scope int(11) NOT NULL COMMENT '数据权限 1、本人;2、本人及下属;3、本部门;4、本部门及下属部门; " \
                   + "5、特定部门6、全部', sort int(11) NULL DEFAULT NULL COMMENT '排序', " \
                   + "status int(11) NULL DEFAULT 1 COMMENT '状态', create_by bigint(20) NOT NULL COMMENT '创建人', " \
                   + "update_by bigint(20) NOT NULL COMMENT '修改人', create_time datetime(0) NOT NULL " \
                   + "DEFAULT CURRENT_TIMESTAMP(0), update_time datetime(0) NOT NULL ON " \
                   + "UPDATE CURRENT_TIMESTAMP(0), del_flag char(1) NOT NULL DEFAULT '0' COMMENT '是否已删除', " \
                   + "PRIMARY KEY ( id ), UNIQUE INDEX role_idx1_role_code2 (role_code)) ENGINE = InnoDB " \
                   + "COMMENT = '系统角色表'; " \
                   + "insert into test_audit_001(id, role_name, role_code, role_desc, role_type, data_scope, " \
                   + "status,create_by, update_by, create_time, update_time, del_flag) values (1, '超级管理员', " \
                   + "'ROLE_ADMIN', NULL, 1, 6, 1, 1, 1, '2021-12-17 00:00:00', '2021-12-17 00:00:00', '0'); " \
                   + "update test_audit_001 set role_name='ado', role_type=1, sort=0, " \
                   + "update_time='2022-12-07 13:10:20', update_by=1 WHERE id=1 AND del_flag=0; " \
                   + "delete from test_audit_001; " \
                   + "select length1(a) as la from (SELECT deptno, AVG(sal) AS avgsal FROM emp GROUP BY deptno) " \
                   + "t2 where b in (select hex(t3.b) from t3 where c=10 having t3.a>1 order by t3.c limit 10); "
        test_order = "python3 " + self.python_file_name + " " + test_so_file_name + " \"" + test_sql + "\""
        self.log.info(test_order)
        source_order = "source " + macro.DB_ENV_PATH + ";"
        self.log.info(source_order)
        test_result = self.db_user_node.sh(source_order + test_order).result()
        self.log.info(test_result)
        self.assertIn('None', test_result.splitlines()[-2], "执行失败")
        self.assertIn('True', test_result.splitlines()[-1], "执行失败")

    def tearDown(self):
        text = '----step2: 清理环境 expect: 成功----'
        self.log.info(text)
        drop_python_order = "rm -rf " + self.python_file_name
        drop_python_result = self.db_user_node.sh(drop_python_order).result()
        self.log.info(drop_python_result)
        self.assertNotIn('ERROR', drop_python_result, "执行失败")
        self.log.info('==Opengauss_Function_Grammar_Audit_Case001完成==')
